package avicit.bdp.dds.server.worker.task.sparkflow;


import avicit.bdp.common.dto.CalculateEngineConf;
import avicit.bdp.common.utils.ConfigUtils;
import avicit.bdp.common.utils.SpringApplicationContext;
import avicit.bdp.core.util.json.JSONUtils;
import avicit.bdp.dds.api.service.LoggerService;
import avicit.bdp.dds.api.service.ProcessResourceConfigService;
import avicit.bdp.dds.api.service.TaskLogService;
import avicit.bdp.dds.api.service.TaskNodesService;
import avicit.bdp.dds.common.Constants;
import avicit.bdp.dds.common.thread.Stopper;
import avicit.bdp.dds.common.utils.YarnUtils;
import avicit.bdp.dds.common.utils.placeholder.BusinessTimeUtils;
import avicit.bdp.dds.common.utils.webhdfs.KerberosWebYarnUtils;
import avicit.bdp.dds.dao.entity.ProcessDefinition;
import avicit.bdp.dds.dao.entity.ProcessInstance;
import avicit.bdp.dds.dao.entity.ProcessResourceSettings;
import avicit.bdp.dds.dao.entity.TaskInstance;
import avicit.bdp.dds.dao.entity.TaskLog;
import avicit.bdp.dds.dao.entity.TaskNodes;
import avicit.bdp.dds.dispatch.enums.ExecutionStatus;
import avicit.bdp.dds.dispatch.enums.TaskType;
import avicit.bdp.dds.dispatch.model.ProcessData;
import avicit.bdp.dds.dispatch.model.TaskNode;
import avicit.bdp.dds.dispatch.task.AbstractParameters;
import avicit.bdp.dds.server.entity.TaskExecutionContext;
import avicit.bdp.dds.server.worker.task.AbstractTask;
import avicit.bdp.dds.server.worker.task.sparkflow.utils.FlowConvertUtils;
import avicit.bdp.dds.server.worker.task.sparkflow.utils.FlowUtils;
import avicit.bdp.dds.server.worker.task.sparkflow.utils.RunModeType;
import avicit.bdp.dds.server.worker.task.sparkflow.utils.ThirdFlowInfoStopVo;
import avicit.bdp.dds.server.worker.task.sparkflow.utils.ThirdFlowInfoStopsVo;
import avicit.bdp.dds.server.worker.task.sparkflow.utils.ThirdFlowInfoVo;
import avicit.bdp.dds.service.process.ProcessService;
import avicit.platform6.commons.utils.ComUtil;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;


/**
 * sparkflow task
 */
public class SparkFlowTask extends AbstractTask {

    private final ProcessService processService;

    /**
     * taskExecutionContext
     */
    private TaskExecutionContext taskExecutionContext;

    private ProcessResourceConfigService processResourceConfigService;

    private TaskLogService taskLogService;

    private TaskNodesService taskNodesService;

    private LoggerService loggerService;

    private ProcessInstance instance;

    private ProcessDefinition definition;

    private Map<String, String> taskInstanceMap = new HashMap<>();

    /**
     * 节点自定义组件Map
     */
    private Map<String, Map<String, String>> nodesMap = new HashMap<>();

    /**
     * constructor
     *
     * @param taskExecutionContext taskExecutionContext
     * @param logger               logger
     */
    public SparkFlowTask(TaskExecutionContext taskExecutionContext, Logger logger) {
        super(taskExecutionContext, logger);
        this.taskExecutionContext = taskExecutionContext;
        this.processService = SpringApplicationContext.getBean(ProcessService.class);
        this.processResourceConfigService = SpringApplicationContext.getBean(ProcessResourceConfigService.class);
        this.taskLogService = SpringApplicationContext.getBean(TaskLogService.class);
        this.taskNodesService = SpringApplicationContext.getBean(TaskNodesService.class);
        this.loggerService = SpringApplicationContext.getBean(LoggerService.class);
    }

    @Override
    public void init() {
        logger.info("sparkflow task json {}", taskExecutionContext.getTaskJson());
        instance = processService.findProcessInstanceById(taskExecutionContext.getProcessInstanceId());
        //初始化日志记录
        try {
            boolean existTaskLog = taskLogService.isExistTaskLog(instance.getId());
            if (!existTaskLog) {
                taskLogService.insertTaskLog(instance.getId(), "");
            }
        } catch (Exception e) {

        }
        //handleLog("开始初始化任务");

        ProcessData processData = JSONUtils.parseObject(instance.getProcessInstanceJson(), ProcessData.class);
        List<TaskNode> taskNodes = processData.getTasks();
        for (TaskNode taskNode : taskNodes) {
            TaskInstance taskInstance = new TaskInstance();
            taskInstance.setId(ComUtil.getId());
            taskInstance.setName(taskNode.getName());
            taskInstance.setTaskType(TaskType.SPARKFLOW.name());
            taskInstance.setProcessDefinitionId(instance.getProcessDefinitionId());
            taskInstance.setProcessInstanceId(instance.getId());
            taskInstance.setTaskJson(JSONUtils.toJson(taskNode));
            taskInstance.setState(ExecutionStatus.SUBMITTED_SUCCESS.getCode());
            taskInstance.setSubmitTime(new Date());
            taskInstance.setHost(taskExecutionContext.getHost());
            taskInstance.setLogPath(taskExecutionContext.getLogPath());
            taskInstance.setExecutePath(taskExecutionContext.getExecutePath());
            taskInstance.setRetryTimes(1);
            taskInstance.setAlertFlag(0);
            taskInstance.setFlag(1);
            taskInstance.setRetryInterval(1);
            taskInstance.setMaxRetryTimes(0);
            taskInstance.setTaskInstancePriority(2);
            taskInstance.setExecutorId(taskExecutionContext.getExecutorId());
            taskInstance.setNodeType(taskNode.getNodeType());
            processService.createTaskInstance(taskInstance);
            //缓存节点名称和id
            taskInstanceMap.put(taskNode.getName(), taskInstance.getId());
        }

        //初始化节点自定义组件
        definition = processService.getProcessDefinitionByProcessInstanceId(taskExecutionContext.getProcessInstanceId());

        //设置process项目ID
        instance.setProjectId(definition.getProjectId());
        List<TaskNodes> nodesList = taskNodesService.getProcessNodeList();
        for (TaskNodes nodes : nodesList) {
            String params = nodes.getParams();
            if (StringUtils.isBlank(params)) {
                continue;
            }
            Map<String, Object> paramsMap = JSONUtils.toObjectMap(params);
            if (paramsMap == null) {
                continue;
            }
            List<Map<String, Object>> propList = (List<Map<String, Object>>) paramsMap.getOrDefault("properties", new ArrayList<>());

            /*
             * 需要参数转换的自定义组件MAP
             * KEY: 组件name值
             * value: 组件component值
             */
            Map<String, String> nodeComponentMap = new HashMap<>(16);
            for (Map<String, Object> propMap : propList) {
                //需要参数转换的自定义组件
                String component = MapUtils.getString(propMap, "component", "");
                if (org.apache.commons.lang3.StringUtils.equalsAnyIgnoreCase(component,
                        "UserTableId", "UserTableName", "UserDataSourceId",
                        "UserResourceFileSelect", "UserResourceSave", "UserDataFileSelect",
                        "UserFileSelect"
                )) {

                    String name = MapUtils.getString(propMap, "name");
                    nodeComponentMap.put(name, component);
                }
            }

            if (!nodeComponentMap.isEmpty()) {
                nodesMap.put(nodes.getId(), nodeComponentMap);
            }
        }
    }

    @Override
    public void handle() throws Exception {
        //处理执行成功后数据
        List<String> postDataList = new ArrayList<>();
        try {
            //  construct process
            Map<String, String> paramsMap = taskExecutionContext.getDefinedParams();
            String checkpoint = MapUtils.getString(paramsMap, "checkpoint", "");
            String runMode = MapUtils.getString(paramsMap, "runMode", "");
            if (StringUtils.isBlank(runMode)) {
                if (definition.getReleaseState() == 1) {
                    runMode = "run";
                } else {
                    runMode = "debug";
                }
            }
            //增加内置日期参数
            Map<String, String> timeParams = BusinessTimeUtils.getBusinessTime(taskExecutionContext.getCmdTypeIfComplement(), taskExecutionContext.getScheduleTime());
            paramsMap.putAll(timeParams);
            //处理启动命令中传入的参数
            String commandParam = instance.getCommandParam();
            if (StringUtils.isNotBlank(commandParam)) {
                Map<String, String> commandParamMap = JSONUtils.toMap(commandParam);
                for (String key : commandParamMap.keySet()) {
                    String val = commandParamMap.get(key);
                    // 如果值不为空，则替换全局参数里的设置
                    if (StringUtils.isNotBlank(val)) {
                        paramsMap.put(key, val);
                    }
                }
            }
            //资源配置
            List<ProcessResourceSettings> prList = processResourceConfigService.getByProcessDefinitionId(taskExecutionContext.getProcessDefineId());
            Map<String, String> redisConfig = processResourceConfigService.getRedisConfig();

            String json = FlowConvertUtils.processToJson(instance, nodesMap, checkpoint,
                    RunModeType.selectGender(runMode), processService, postDataList, taskInstanceMap,
                    taskExecutionContext.getCalculateEngineConf(), paramsMap, definition.getDatasourceId(), prList, redisConfig);

            logger.info("sparkFlowJson: \n" + json + "\n");

            //handleLog("开始提交spark任务");
            //handleLog(json);
            String appid = FlowUtils.startFlow(json);
            logger.info("appId: " + appid + "\n");
            if (StringUtils.isBlank(appid) || "-1".equals(appid)) {
                logger.error("{}", "流程启动失败");
                //日志
                handleErrorLog("任务启动失败");
                handleLog(json);
                instance.setState(ExecutionStatus.FAILURE.getCode());
                processService.updateProcessInstance(instance);
                setExitStatusCode(Constants.EXIT_CODE_FAILURE);

                handleStopWebsocketInfo(instance.getProcessDefinitionId(), instance.getId(), ExecutionStatus.FAILURE.getCode());

            } else {
                //开始运行时间
                long startRunTime = System.currentTimeMillis();
                //保存appid到instance表的extra_params
                Map<String, Object> exParams = new HashMap<>(16);
                exParams.put("appid", appid);
                ProcessInstance processInstance = new ProcessInstance();
                processInstance.setId(taskExecutionContext.getProcessInstanceId());
                processInstance.setExtraParams(JSONUtils.toJson(exParams));
                processService.updateProcessInstanceByPrimaryKeySelective(processInstance);
                processService.updateTaskAppIdByProcessInstanceId(taskExecutionContext.getProcessInstanceId(), appid);

                handleStopWebsocketInfo(instance.getProcessDefinitionId(), instance.getId(), ExecutionStatus.RUNNING_EXEUTION.getCode());

                ThirdFlowInfoVo flowInfo = null;

                //提交成功，处理节点运行状态
                while (Stopper.isRunning()) {
                    try {
                        //判断任务停止后，自动退出线程
                        instance = processService.findProcessInstanceById(taskExecutionContext.getProcessInstanceId());
                        if (ExecutionStatus.READY_STOP.getCode() == instance.getState().intValue() || ExecutionStatus.STOP.getCode() == instance.getState().intValue()) {
                            killYarnApp(appid);
                            instance.setState(ExecutionStatus.STOP.getCode());
                            processService.updateProcessInstance(instance);
                            handleStopTask(flowInfo);
                            handleLog("任务停止");
                            handleStopWebsocketInfo(instance.getProcessDefinitionId(), instance.getId(), ExecutionStatus.STOP.getCode());
                            break;
                        } else if (instance.isProcessInstanceStop()) {
                            //运行任务状态的变化由spark运行框架来更新
                            break;
                        }

                        //判断是否超出系统设置的最长超时时间(小时), 如果为0，则不进行控制，
                        long maxRunTime = ConfigUtils.getInstance().getLong("dds.task.maxRunTime", 72);
                        //转换成毫秒
                        maxRunTime = maxRunTime * 60 * 60 * 1000;
                        long endRunTime = System.currentTimeMillis();
                        if (maxRunTime != 0 && (endRunTime - startRunTime) > maxRunTime) {
                            killYarnApp(appid);
                            instance.setState(ExecutionStatus.FAILURE.getCode());
                            processService.updateProcessInstance(instance);
                            handleStopWebsocketInfo(instance.getProcessDefinitionId(), instance.getId(), ExecutionStatus.FAILURE.getCode());
                            break;
                        }

                        Thread.sleep(1000);

                    } catch (Exception e) {
                        logger.error("dispatcher task error", e);
                        //日志
                        handleErrorLog("流程运行失败:\n" + getExceptionInfo(e));
                        handleStopWebsocketInfo(instance.getProcessDefinitionId(), instance.getId(), ExecutionStatus.FAILURE.getCode());
                        break;
                    }
                }

                //处理任务状态
                ExecutionStatus status = ExecutionStatus.of(instance.getState());
                if (status.typeIsSuccess()) {
                    setExitStatusCode(Constants.EXIT_CODE_SUCCESS);
                    //handleLog("任务运行成功");
                } else {
                    setExitStatusCode(Constants.EXIT_CODE_FAILURE);
                    handleLog("任务运行失败");
                }
            }

            setAppIds(appid);
        } catch (Exception e) {
            logger.error("spark flow task failure", e);
            //日志
            handleErrorLog("任务运行失败:\n" + getExceptionInfo(e));

            setExitStatusCode(Constants.EXIT_CODE_FAILURE);
            handleStopWebsocketInfo(instance.getProcessDefinitionId(), instance.getId(), ExecutionStatus.FAILURE.getCode());
            throw e;
        }
    }

    private void killYarnApp(String appId) throws Exception {
        CalculateEngineConf conf = taskExecutionContext.getCalculateEngineConf();
        if (conf != null && Constants.KERBEROS.equals(conf.getAuthType())) {
            KerberosWebYarnUtils utils = new KerberosWebYarnUtils(conf.getYarnAddress(), conf.getKrb5Path(),
                    conf.getKeytabPath(), conf.getKerberosUser());
            utils.killApp(appId);
        } else {
            YarnUtils.killYarnApp(conf, appId);
            //FlowUtils.stopFlow(appid);
        }
    }

    private void handleStopWebsocketInfo(String processDefinitionId, String processId, Integer runningState) {
        loggerService.sendWebsocketTaskStatus(processDefinitionId, processId, runningState);
    }

    private void handleStopTask(ThirdFlowInfoVo flowInfo) {
        if (flowInfo != null) {
            List<ThirdFlowInfoStopsVo> stops = flowInfo.getStops();
            for (ThirdFlowInfoStopsVo stop : stops) {
                ThirdFlowInfoStopVo vo = stop.getStop();
                //如果状态发生变化，则进行更新数据库
                if ("STARTED".equalsIgnoreCase(vo.getState())) {
                    String id = taskInstanceMap.get(vo.getName());
                    TaskInstance taskInstance = new TaskInstance();
                    taskInstance.setId(id);
                    taskInstance.setEndTime(new Date());
                    //进度
                    taskInstance.setState(ExecutionStatus.KILL.getCode());
                    processService.updateTaskInstanceByPrimaryKeySelective(taskInstance);
                }
            }
        }
    }

    private void handleLog(String content) {
        TaskLog taskLog = new TaskLog();
        taskLog.setId(instance.getId());
        taskLog.setTaskName("系统");
        taskLog.setContents(content);
        loggerService.insertTaskLog(taskLog);
    }

    private void handleErrorLog(String content) {
        TaskLog taskLog = new TaskLog();
        taskLog.setId(instance.getId());
        taskLog.setTaskName("系统");
        taskLog.setContents(content);
        loggerService.insertTaskLog(taskLog);
    }

    @Override
    public AbstractParameters getParameters() {
        return null;
    }

    public String getExceptionInfo(Exception ex) {
        StringBuilder messages = new StringBuilder();
        StackTraceElement[] message = ex.getStackTrace();
        for (StackTraceElement element : message) {
            messages.append("\t").append(element.toString()).append("\n");
        }
        return messages.toString();
    }

}
